package com.bw.gmall.realtime.dwd.db.solit.app;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.common.base.BaseApp;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.util.DateFormatUtil;
import com.bw.gmall.realtime.common.util.FlinkSinkUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.sink.KafkaSinkBuilder;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SideOutputDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.apache.kafka.clients.producer.ProducerConfig;


/**
 * 日志分流
 * 需要启动的进程
 *  zk.kfk.flume.DwdBaseLog
 *
 *  kafkaSource:从kafaka主题中读取数据
 *                通过手动维护便宜,保证消费的精准一次
 *
 *  kafkaSink:向kafka主题中写入数据,也可以保证写入的精准一次,需要做如下操作
 *          开启检查点
 *          .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
 *          .setTransactionalIdPrefix("dwd_base_log")
 *          .setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG,15*60*1000+"")
 *          在消费段,需要设置消费的隔离界级别为读已提交 .setProperty(ConsumerConfig.ISOLATION_LEVEL_CONFIG,"read_uncommitted")
 *
 */
public class DwdBaseLog extends BaseApp {
    public static void main(String[] args) {
        new DwdBaseLog().start(10011,4, "dwd_base_log",Constant.TOPIC_LOG);
    }
    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        //TODO 对流中数据类型进行转换  并做简单的ETL
        //定义测输出流标签
        OutputTag<String> dirtyTag = new OutputTag<String>("dirtyTag"){};
        //ETL
        SingleOutputStreamOperator<JSONObject> jsonObjDS = etl(stream, dirtyTag);

//        jsonObjDS.print("标准的json:>>>");
        SideOutputDataStream<String> dirtyDS = jsonObjDS.getSideOutput(dirtyTag);
//        dirtyDS.print("脏数据:>>>");
     //将测输出流中的脏数据写到kafka中

//        KafkaSink<String> kafkaSink = KafkaSink.<String>builder()
//                .setBootstrapServers(Constant.KAFKA_BROKERS)
//                // 反序列化
//                .setRecordSerializer(KafkaRecordSerializationSchema.builder()
//                        .setTopic("dirty_data")
//                        .setValueSerializationSchema(new SimpleStringSchema())
//                        .build())
//                // kafka 精准一次性 怎们 保证 :  当前配置决定是否开启事务,保证写道kafka数据的精准一次
////                .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
//                //设置事务ID的前缀
////                .setTransactionalIdPrefix("dwd_base_log")
//                //设置事物的超时时间  检查点超时时间<事物的超时时间<=事务最大的超时时间
////                .setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG,15*60*1000+"")
//                .build();

//        dirtyDS.sinkTo(kafkaSink);


        KafkaSink<String> dirtyData = FlinkSinkUtil.getKafkaSink("dirty_data");

        //TODO 对新老用户进行修复
        //按照设备id进行分组
        KeyedStream<JSONObject, String> keyDS = jsonObjDS.keyBy(jsonObject -> jsonObject.getJSONObject("common").getString("mid"));
        //使用Flink的状态编程完成修复
        SingleOutputStreamOperator<JSONObject> isNewDS = keyDS.map(new RichMapFunction<JSONObject, JSONObject>() {
            private ValueState<String> lastVisitDateState;

            @Override
            public void open(Configuration parameters) throws Exception {
                //状态声明 以及 状态初始化
                ValueStateDescriptor<String> valueStateDescriptor =
                        new ValueStateDescriptor<>("lastVisitDateState", String.class);

                lastVisitDateState = getRuntimeContext().getState(valueStateDescriptor);
            }

            @Override
            public JSONObject map(JSONObject jsonObject) throws Exception {
                //获取is_new 的值
                String isNew = jsonObject.getJSONObject("common").getString("is_new");
                JSONObject common = jsonObject.getJSONObject("common");
                //从状态中获取首次访问日期
                String lastDate = lastVisitDateState.value();
                //获取当前访问日期
                Long ts = jsonObject.getLong("ts");
                String curVisitDate = DateFormatUtil.tsToDate(ts);
                if ("1".equals(isNew))
                //如果is_new的值为1
                {
                    if (StringUtils.isEmpty(lastDate)) {
                        //如果键控状态为null,认为本次是该访客首次访问App,将日志中ts对应的日期更新到状态中,不对is_new 字段做修改;
                        lastVisitDateState.update(curVisitDate);
                    } else {
                        //如果键控状态部位null,并且首次访问日期不是当日,说明是老访客 将is_new 改为 0
                        if (!lastDate.equals(curVisitDate)) {
                            isNew = "0";
                            common.put("is_new", isNew);
                        }
                    }
                } else {
                    //如果is_new 的值为0
                    //如果 数仓里面 状态 为null
                    if (StringUtils.isEmpty(lastDate)) {
                        //获取昨天的日期存入 状态里(数仓中)
                        String yesterday = DateFormatUtil.tsToDate(ts - 24 * 60 * 60 * 1000);
                        lastVisitDateState.update(yesterday);
                    }
                }
                return jsonObject;
            }
        });

        //TODO 分流 错误日志-错误测输出流 启动日志-启动测输出流 保管入职-曝光测输出流 动作日志-动作测输出流  页面日志-主流
        //定义测输出标签
        OutputTag<String> startTag = new OutputTag<String>("startTag") {};
        OutputTag<String> errTag = new OutputTag<String>("errTag") {};
        OutputTag<String> displayTag = new OutputTag<String>("displayTag") {};
        OutputTag<String> actionTag = new OutputTag<String>("actionTag") {};

        //分流
        SingleOutputStreamOperator<String> pageDS = isNewDS.process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject jsonObject, ProcessFunction<JSONObject, String>.Context ctx, Collector<String> out) throws Exception {
                // ---错误日志---
                JSONObject errJsonObj = jsonObject.getJSONObject("err");
                if (errJsonObj != null) {
                    //将错误日志写道错误测输出流
                    ctx.output(errTag, jsonObject.toJSONString());
                    jsonObject.remove("err");
                }
                JSONObject startJsonObj = jsonObject.getJSONObject("start");
                if (startJsonObj != null) {
                    // ---启动日志---
                    //将启动日志存入到测输出流
                    ctx.output(startTag,jsonObject.toJSONString());

                } else {
                    // ---页面日志---
                    JSONObject commonJsonObj = jsonObject.getJSONObject("common");
                    JSONObject pageJsonObj = jsonObject.getJSONObject("page");
                    Long ts = jsonObject.getLong("ts");

                    // ---曝光日志---
                    JSONArray displayArray = jsonObject.getJSONArray("displays");
                    if (displayArray != null && displayArray.size() > 0) {
                        //遍历当前页面的所有曝光信息
                        for (int i = 0; i < displayArray.size(); i++) {
                            JSONObject displayJsonObj = displayArray.getJSONObject(i);
                            //自定义个新的 JSON 对象,用于封装遍历出来的 曝光数据
                            JSONObject newDisplayJsonObj = new JSONObject();
                            newDisplayJsonObj.put("common", commonJsonObj);
                            newDisplayJsonObj.put("page", pageJsonObj);
                            newDisplayJsonObj.put("display", displayJsonObj);
                            newDisplayJsonObj.put("ts", ts);

                            //将曝光日志 写入曝光测输出流
                            ctx.output(displayTag, newDisplayJsonObj.toJSONString());
                        }
                        jsonObject.remove("displays");
                    }

                    // ---动作日志---
                    JSONArray actionsArray = jsonObject.getJSONArray("actions");
                    if (actionsArray != null && actionsArray.size() > 0) {
                        //遍历当前页面的所有曝光信息
                        for (int i = 0; i < actionsArray.size(); i++) {
                            JSONObject actionJsonObj = actionsArray.getJSONObject(i);
                            //自定义个新的 JSON 对象,用于封装遍历出来的 曝光数据
                            JSONObject newActionJsonObj = new JSONObject();
                            newActionJsonObj.put("common", commonJsonObj);
                            newActionJsonObj.put("page", pageJsonObj);
                            newActionJsonObj.put("action", actionJsonObj);
                            newActionJsonObj.put("ts", ts);

                            //将动作日志 写入曝光测输出流
                            ctx.output(actionTag, newActionJsonObj.toJSONString());
                        }
                        jsonObject.remove("actions");
                    }

                    //页面日志 存入主流中
                    out.collect(jsonObject.toJSONString());

                }
            }
        });

        SideOutputDataStream<String> errDS = pageDS.getSideOutput(errTag);
        SideOutputDataStream<String> startDS = pageDS.getSideOutput(startTag);
        SideOutputDataStream<String> displayDS = pageDS.getSideOutput(displayTag);
        SideOutputDataStream<String> actionDS = pageDS.getSideOutput(actionTag);

        pageDS.print("页面>>>");
        errDS.print("错误>>>");
        startDS.print("启动>>>");
        displayDS.print("曝光>>>");
        actionDS.print("动作>>>");

        //TODO 将不同流的数据写到kafka的不同主题中
        pageDS.sinkTo(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_PAGE));
        errDS.sinkTo(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_ERR));
        startDS.sinkTo(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_START));
        displayDS.sinkTo(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_DISPLAY));
        actionDS.sinkTo(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_ACTION));

    }




    private static SingleOutputStreamOperator<JSONObject> etl(DataStreamSource<String> stream, OutputTag<String> dirtyTag) {
        return stream.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String jsonStr, ProcessFunction<String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                        try {
                            JSONObject jsonObj = JSONObject.parseObject(jsonStr);
                            // 如果转换时 没有发生异常,说明是标准的 json 将数据传递下游
                            out.collect(jsonObj);
                        } catch (Exception e) {
                            //如果转换是 发生异常 说明不是标准的 json 将数据放入到 测流中
                            ctx.output(dirtyTag, jsonStr);
                        }

                    }
                }
        );
    }
}
